package iot.hbase;

import hbase.template.HBaseAsyncTemplate;
import iot.entity.Registration;
import iot.hbase.codec.RegistrationCodec;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.FirstKeyOnlyFilter;
import org.apache.hadoop.hbase.filter.PageFilter;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Service;

import java.util.Date;
import java.util.List;
import java.util.concurrent.CompletableFuture;

/**
 * Created by dave on 18-7-21 上午10:11.
 */
@Service
@ConditionalOnExpression("${hbase.async:false}")
@ConditionalOnProperty(name = "dao.type", havingValue = "hbase")
public class RegistrationAsyncDao {
    private static final String TABLE = "registration";
    @Autowired HBaseAsyncTemplate template;
    private RegistrationCodec codec = new RegistrationCodec();

    public CompletableFuture<Void> insert(Registration reg) {
        return template.put(TABLE, reg, codec);
    }

    public void batchInsert(List<Registration> list) {
        template.batchInsert(TABLE, list, codec);
    }

    public CompletableFuture<List<Registration>> getList(Date start, Date end, int limit) {
        Scan scan = new Scan();
        scan.withStartRow(Bytes.toBytes(start.getTime()));
        scan.withStopRow(Bytes.toBytes(end.getTime()));
        scan.setFilter(new PageFilter(limit));
        return template.find(TABLE, scan, codec);
    }

    public CompletableFuture<List<Registration>> getListAfter(Date start, int limit) {
        Scan scan = new Scan();
        scan.withStartRow(Bytes.toBytes(start.getTime()));
        scan.setFilter(new PageFilter(limit));
        return template.find(TABLE, scan, codec);
    }

    public CompletableFuture<List<Registration>> getListBefore(Date end, int limit) {
        Scan scan = new Scan();
        scan.withStartRow(Bytes.toBytes(end.getTime()), true);
        scan.setFilter(new PageFilter(limit));
        scan.setReversed(true);
        return template.find(TABLE, scan, codec);
    }

    public long rowCount(Date start, Date end) {
        Scan scan = new Scan();
        scan.withStartRow(Bytes.toBytes(start.getTime()));
        scan.withStopRow(Bytes.toBytes(end.getTime()));
        scan.setFilter(new FirstKeyOnlyFilter());
        return template.rowCount(TABLE, scan);
    }

    public long rowCount() {
        Scan scan = new Scan();
        scan.setFilter(new FirstKeyOnlyFilter());
        return template.rowCount(TABLE, scan);
    }
}
